package org.apache.beam.examples.jdbc;

import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import org.apache.beam.examples.Models.*;
import org.apache.beam.examples.Models.enums.QualityType;
import org.apache.beam.examples.PTransforms.QualityRuleMain;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.jdbc.JdbcIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.values.PCollection;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * 本类主要用于编排数据质量管理规则的内容，尝试使用这部分内容完成数据库中的大数据量质量检查
 * <ol>
 * <li> 读取应用配置，将配置数据从文件/数据库中读出 </li>
 * <li> 根据规则配置完成规则的编排与处理过程 </li>
 * <li> 数据对象为待处理的数据集以及待执行的规则
 *  <ol>
 *      <li> 因为处理的是一行数据，所以，入栈内容就是通过规则以及行数据  </li>
 *      <li> 运行规则  </li>
 *      <li> 拼接规则结果 </li>
 *      <li> 插入规则结果数据库 </li>
 *  </ol>
 * </li>
 * </ol>
 */
public class BeamH2DataSparkExample {

    static final Log log = LogFactory.get();

    public static void main(String[] args) {


        NounNode certType = new NounNode();

        certType.setName("证件类型");
        certType.setCode("cerType");
        certType.childrenAdd(new NounNode("10", "证件类型1", "信息说明", "sys", "", certType, null));
        certType.childrenAdd(new NounNode("20", "证件类型2", "信息说明", "sys", "", certType, null));
        certType.childrenAdd(new NounNode("30", "证件类型3", "信息说明", "sys", "", certType, null));

        NounNode userState = new NounNode();
        userState.setName("userState");
        userState.setCode("userState");
        // RESIGN ONJOB
        userState.childrenAdd(new NounNode("RESIGN", "离职", "信息说明", "sys", "", userState, null));
        userState.childrenAdd(new NounNode("ONJOB", "在职", "信息说明", "sys", "", userState, null));


        List<QualityRuleDetailModel> qualityRuleDetailModels = new ArrayList<>();


        qualityRuleDetailModels.add(new QualityRuleDetailModel("COLID001", "ID001", "account", "用户编号", List.of(QualityType.hasNull), null));
        // CERTTYPE
        qualityRuleDetailModels.add(new QualityRuleDetailModel("COLID002", "ID001", "CERTTYPE", "证件类型", List.of(QualityType.hasNull,QualityType.OutNounItem), certType));
        // USERSTATE
        qualityRuleDetailModels.add(new QualityRuleDetailModel("COLID003", "ID001", "USERSTATE", "用户状态", List.of(QualityType.hasNull,QualityType.OutNounItem), userState));


        // 构筑校验规则对象
        QualityRuleMainModel model = new QualityRuleMainModel();
        model.setQualityId("ID001");
        model.setQualityDescription("测试用规则说明。");
        model.setQualityName("测试用规则名称");
        model.setQualityRuleDetailModels(qualityRuleDetailModels);

        PipelineOptions options = PipelineOptionsFactory.fromArgs(args).withValidation().create();
        Pipeline pipeline = Pipeline.create(options);

        String query = "SELECT * FROM sys_user";


        pipeline.apply(JdbcIO.<ArrayList<QualityRuleColData>>read()
                .withDataSourceConfiguration(JdbcIO.DataSourceConfiguration.create(
                                H2DataBaseInit.driver, H2DataBaseInit.url)
                        .withUsername(H2DataBaseInit.user)
                        .withPassword(H2DataBaseInit.password))
                .withQuery(query) // TODO 需要在调整。
                .withRowMapper(new JdbcIO.RowMapper<ArrayList<QualityRuleColData>>() {
                    public ArrayList<QualityRuleColData> mapRow(ResultSet resultSet) throws SQLException {
                        log.info("获取到数据了:{}", resultSet.toString());
                        String account = resultSet.getString(1);
                        String userstate = resultSet.getString("USERSTATE");
                        String certType = resultSet.getString("CERTTYPE");
                        Map<String ,Object> dataMap = Map.of("ACCOUNT", account, "USERSTATE", userstate, "CERTTYPE", certType);
                        ArrayList<QualityRuleColData> qualityRuleColDataList = new ArrayList<>();
                        for (QualityRuleDetailModel qualityRuleDetailModel : model.getQualityRuleDetailModels()) {
                            qualityRuleColDataList.add(new QualityRuleColData(dataMap,qualityRuleDetailModel));
                        }
                        return qualityRuleColDataList;
                    }
                })
        ).apply(new QualityRuleMain());

        pipeline.run().waitUntilFinish();
    }
}
